package com.shiguiwu.springboot3.kafka.config;

import com.shiguiwu.springboot3.kafka.listener.MyProducerListener;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

/**
 * @description: 配置
 * @author: stone
 * @date: Created by 2023/11/13 17:40
 * @version: 1.0.0
 * @pakeage: com.shiguiwu.springboot3.kafka.config
 */
@Configuration
public class KafkaConfig {
    /**
     * 注入统一的消息拦截监听
     * @return
     */
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory) {
        KafkaTemplate<String, Object> kafkaTemplate = new KafkaTemplate<>(producerFactory);
        kafkaTemplate.setProducerListener(new MyProducerListener());
        return kafkaTemplate;
    }

    @Autowired
    ConsumerFactory consumerFactory;

    //消息过滤器
    //@Bean
    public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        //被过滤的消息将被丢弃
        factory.setAckDiscarded(true);
        //消息过滤策略
        factory.setRecordFilterStrategy(consumerRecord -> {
            if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {
                return false;
            }
            return true;
        });
        return factory;
    }


    @Bean
    public NewTopic test() {
        return TopicBuilder
                .name("shigwTopic")
                .replicas(1)
                .partitions(3)
                .build();
    }


}
